package com.heima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.behavior.dto.ReadBehaviorDto;
import com.heima.behavior.dto.UpdateArticleMessage;
import com.heima.behavior.entity.ApLikesBehavior;
import com.heima.behavior.entity.ApReadBehavior;
import com.heima.behavior.service.IReadService;
import com.heima.common.dto.ResponseResult;
import com.heima.common.dto.User;
import com.heima.common.util.AppThreadLocalUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;

@Service
public class ReadServiceImpl implements IReadService {

    @Autowired
    MongoTemplate mongoTemplate; // mongo数据库

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate; // fakfa 消息中间件 发送消息
    @Value("${topic.read_behavior}")
    String ReadTopic; // 发送消息到 哪个话题

    /**
     * 阅读行为记录
     * @param dto
     * @return
     */
    @Override
    public ResponseResult saveRead(ReadBehaviorDto dto) {
        //1. 构建 APP阅读行为表 对象
        ApReadBehavior behavior = new ApReadBehavior();
        //2. 用线程取用户信息
        User user = AppThreadLocalUtil.get();
        //3. 取到说明是用户登录, 取不到说明是游客登录, 用设备id
        if (user != null){
            // 写入用户id
            behavior.setUserId(user.getUserId());
        } else {
            // 写入设备id, 从前端出入
            behavior.setEquipmentId(dto.getEquipmentId());
        }
        // 写入文章、动态、评论等ID, 从前端传入
        behavior.setArticleId(dto.getArticleId());
        // 写入阅读次数 第一次为 1
        behavior.setCount(1);
        // 写入阅读时间单位, 从前端传入
        behavior.setReadDuration(dto.getReadDuration());
        // 写入阅读百分比, 从前端传入
        behavior.setPercentage(dto.getPercentage());
        // 写入文章加载时间, 从前端传入
        behavior.setLoadDuration(dto.getLoadDuration());
        // 写入创建时间
        behavior.setCreatedTime(new Date());
        //4. 查询当前用户或者设备对当前文章是否已经有历史记录
        Query query = new Query();
        // 文章、动态、评论等ID条件
        query.addCriteria(Criteria.where("articleId").is(dto.getArticleId()));
        // 判断是不是用户登录进行点赞, 不是用户, 就设备id,          是的话, 就用户id
        if (user == null){
            // 非用户登录, 设备id 条件
            query.addCriteria(Criteria.where("equipmentId").is(dto.getEquipmentId()));
        } else {
            // 用户登录, 用户id 条件
            query.addCriteria(Criteria.where("userId").is(user.getUserId()));
        }
        // 调用mongo进行条件查询
        ApReadBehavior one = mongoTemplate.findOne(query, ApReadBehavior.class);
        // 如果没有查到记录, 说明没浏览过, 要啊进行新增
        if (one == null){
            // 调用mongo进行新增, 保存
                mongoTemplate.save(behavior);
            }  else {
            // 如果查到了记录, 更新记录
            // 每次将阅读记录+1   1秒内 有100个用户同时阅读  加入数据库原本阅读量是1  应该更新到101
            // one.setCount(one.getCount()+1); 这种方式不是原子操作
            // 构建更新条件
            Update update = new Update();
            // inc 是一个原子操作, 阅读次数 count = count + 1
            update.inc("count");
            // 更新阅读时间单位
            update.set("readDuration", dto.getReadDuration());
            // 更新阅读百分比
            update.set("percentage", dto.getPercentage());
            // 更新文章加载时间
            update.set("loadDuration", dto.getLoadDuration());
            //更新时间
            update.set("updatedTime", new Date());
            // 调用mongo数据库进行修改
            mongoTemplate.updateFirst(query, update, ApReadBehavior.class);
        }
        //5. 发送消息到kafka
        // 构建消息对象
        UpdateArticleMessage message = new UpdateArticleMessage();
        // 写入操作类型 0 阅读 1 点赞 2 评论 3 收藏
        message.setType(0);
        // 写入文章ID, 从前端传入
        message.setArticleId(dto.getArticleId());
        // 写入修改数据的增量，可为正负
        message.setAdd(1);
        // 定义value  转json串
        String value = JSON.toJSONString(message);
        // 调用kafka 发送消息
        kafkaTemplate.send(ReadTopic, value);
        //last  返回成功标志
        return ResponseResult.okResult();
    }
}
